package com.we.risk.phoneBook.dimuser;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import com.alibaba.fastjson.JSONObject;
import com.we.risk.phoneRecord.common.Stage2GlobalSetting;
import com.weicai.caesar.CaesarUtil;
import org.apache.commons.io.FileUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.util.HashMap;
import java.util.List;

import static com.we.risk.phoneRecord.common.Stage2GlobalSetting.PIDAREAFILENAME;

public class TableUserPidMem {
    public static final String KFKTABLE = "fdm.hhl01_user_pid";
    public static final String QUERYSQL =
            "select uid, pid, updated_at "
                    + " from "
                    + KFKTABLE
                    + " where deleted_at is null "
                    + "and updated_at is not null "
                    + "and pid is not null";
    public static final int PIDLENGTH = 6;
    public static final String INPUT_KEY_BY = "uid";
    public static final String SPLITWORD = "#";

    public long uid;
    public String pid;
    public int updated_at;

    public TableUserPidMem() {}

    public TableUserPidMem(long uid, String pid, int updated_at) {
        this.uid = uid;
        this.pid = pid;
        this.updated_at = updated_at;
    }

    public static Logger LOG = LoggerFactory.getLogger(TableUserPidMem.class);

    public static TableUserPidMem init(long uid) {
        return new TableUserPidMem(uid, "", -1);
    }

    public static SingleOutputStreamOperator<String> getUserPidProcessedStream(
            StreamTableEnvironment tableEnv, StreamExecutionEnvironment env) {
        Table tableRes = tableEnv.sqlQuery(TableUserPidMem.QUERYSQL);
        DataStream<Tuple2<Boolean, TableUserPidMem>> inputStream =
                tableEnv.toRetractStream(tableRes, TableUserPidMem.class);

        env.registerCachedFile(Stage2GlobalSetting.PIDAREDATA, Stage2GlobalSetting.PIDAREAFILENAME);

        SingleOutputStreamOperator<String> out =
                inputStream
                        .filter(
                                new FilterFunction<Tuple2<Boolean, TableUserPidMem>>() {
                                    @Override
                                    public boolean filter(Tuple2<Boolean, TableUserPidMem> value)
                                            throws Exception {
                                        return value.f0;
                                    }
                                })
                        .uid("TableUserPid-filter")
                        .keyBy(new TableUserPidKeySelectorFunc())
                        .flatMap(new TableUserPidFlatMapFunc());

        return out;
    }

    private static class TableUserPidKeySelectorFunc
            implements KeySelector<Tuple2<Boolean, TableUserPidMem>, Long> {
        @Override
        public Long getKey(Tuple2<Boolean, TableUserPidMem> value) throws Exception {
            return value.f1.getUid();
        }
    }

    private static class TableUserPidFlatMapFunc
            extends RichFlatMapFunction<Tuple2<Boolean, TableUserPidMem>, String> {
        ValueState<TableUserPidMem> tableUserPidVSatate;
        HashMap<String, String> pidAreaMap;

        @Override
        public void open(Configuration parameters) throws Exception {
            tableUserPidVSatate =
                    getRuntimeContext()
                            .getState(
                                    new ValueStateDescriptor<TableUserPidMem>(
                                            "TableUserPidVSatate", TableUserPidMem.class));

            initPidAreaMap();
        }

        @Override
        public void flatMap(Tuple2<Boolean, TableUserPidMem> value, Collector<String> out)
                throws Exception {
            try {
                long uid = value.f1.getUid();
                int updated_at = value.f1.getUpdated_at();
                String pid = value.f1.getPid();
                if (CaesarUtil.isEncrypted(pid)) {
                    pid = CaesarUtil.decode(pid);
                }
                if (pid.length() >= PIDLENGTH) {
                    pid = pid.substring(0, PIDLENGTH);
                }

                TableUserPidMem tmpState = tableUserPidVSatate.value();
                TableUserPidMem curSTate = tmpState == null ? TableUserPidMem.init(uid) : tmpState;

                if (updated_at > curSTate.getUpdated_at()) {
                    if (!pid.equals(curSTate.pid)) {
                        String res = pidAreaMap.get(pid);
                        if (res != null) {
                            String[] splits = res.split(SPLITWORD);
                            if (splits.length == 3) {
                                JSONObject json = new JSONObject();
                                json.put("uid", uid);
                                json.put("pid", pid);
                                json.put("pid_prov", splits[0].equals("null") ? "" : splits[0]);
                                json.put("pid_city", splits[1].equals("null") ? "" : splits[1]);
                                json.put("pid_area", splits[2].equals("null") ? "" : splits[2]);

                                out.collect(json.toString());
                            }
                        } else {
                            /** 未关联到维表数据不向下输出 */
                            LOG.warn("UserJPidFunc ProcData pid: " + pid);
                        }
                        curSTate.updated_at = updated_at;
                        curSTate.pid = pid;
                        tableUserPidVSatate.update(curSTate);
                    }
                }
            } catch (Exception e) {
                LOG.error("TableUserPid flatMap Exception: " + e.toString());
                LOG.error("TableUserPid flatMap Exception input: " + value);
                e.printStackTrace();
            }
        }

        private void initPidAreaMap() throws IOException {
            pidAreaMap = new HashMap<>();
            File pidArea = getRuntimeContext().getDistributedCache().getFile(PIDAREAFILENAME);
            List<String> pidAreaLines = FileUtils.readLines(pidArea);
            for (String line : pidAreaLines) {
                String[] splits = line.split(",");
                if (splits.length == 4) {
                    String pid = splits[0];
                    String province = splits[1];
                    String city = splits[2];
                    String county = splits[3];
                    /** 维表异常数据处理 */
                    if (province == null || province.equals("")) {
                        province = "null";
                    }

                    if (city == null || city.equals("")) {
                        city = "null";
                    }

                    if (county == null || county.equals("")) {
                        county = "null";
                    }

                    String mapKey = pid;
                    String mapVal = province + SPLITWORD + city + SPLITWORD + county;
                    pidAreaMap.put(mapKey, mapVal);
                } else {
                    LOG.error("DimPhoneRecordLabel initV1CleanMap error read line: " + line);
                }
            }
        }
    }

    public long getUid() {
        return uid;
    }

    public int getUpdated_at() {
        return updated_at;
    }

    public String getPid() {
        return pid;
    }
}
